package node;

import java.net.InetAddress;
import java.util.concurrent.LinkedBlockingQueue;
import task.IQueue;
import task.Task;

public class NodeExecThread extends Thread implements NodeExecWorkerListener{

	protected IQueue 		theQueue; 
	protected InetAddress 	discoveryAddress;
	protected int 			discoveryPort;
	protected LinkedBlockingQueue<Runnable>	workerPool;
	protected boolean		pauseThread;
	
	public NodeExecThread(IQueue taskQueue, InetAddress discoveryAddress, int discoveryPort)
	{
		super("NodeExecThread");
		theQueue 				= taskQueue;
		
		this.discoveryAddress 	= discoveryAddress;
		this.discoveryPort 		= discoveryPort; 
		
		workerPool				= new LinkedBlockingQueue<Runnable>();
		
		pauseThread 			= false;
	}
	
	public void setPauseThread(boolean newValue)
	{
		this.pauseThread = newValue;
	}
	
	@Override
	public void run() {
		// TODO Auto-generated method stub
		while(true)
		{
			
			Task currentTask = theQueue.dequeue();
			if(currentTask != null && workerPool.size() < NodeDescriptor.getInstance().getIMaxThreads())
			{
				NodeExecWorker taskWorker = new NodeExecWorker(currentTask, discoveryAddress, discoveryPort);
				taskWorker.workerDelegate = this;
				workerPool.add(taskWorker);
				new Thread(taskWorker, "TaskWorker:"+currentTask.getCommand()).start();
			}
			
			synchronized(this)
			{
				while(this.pauseThread)
				{
					try{
						this.wait();
					} catch(InterruptedException e) {}
				}		
			}

		}
	}
	
	@Override
	public synchronized void workerHasFinished(NodeExecWorker w) {
		// TODO Auto-generated method stub
		this.workerPool.remove(w);
	}
}